package com.vehicle.stock.provider.rocketmq.consumer;

import com.google.common.collect.Maps;
import com.vehicle.common.util.Jsons;
import com.vehicle.stock.api.service.SeckillService;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * Created by yantingbin on 2017/8/11.
 */
@Component
public class PushConsumer {
    private static final Logger LOGGER= LoggerFactory.getLogger(PushConsumer.class);
    private static final DefaultMQPushConsumer CONSUMER = new DefaultMQPushConsumer("please_rename_unique_group_name_5");

    @Autowired
    private SeckillService seckillService;

    @PostConstruct
    public void init() throws MQClientException {
        CONSUMER.setNamesrvAddr("172.16.14.76:9876");
        CONSUMER.subscribe("SeckillTopic", "*");
        CONSUMER.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //wrong time format 2017_0422_221800
        CONSUMER.setConsumeTimestamp(String.valueOf(System.currentTimeMillis()));
        CONSUMER.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                LOGGER.info(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");

                LOGGER.info("body data :{}", new String(msgs.get(0).getBody()));
                Map<String,Object> map= (Map)Jsons.fromJson(new String(msgs.get(0).getBody()));

               if(seckillService.reduceSeckill(Long.valueOf(map.get("seckillId").toString()),new Date())){
                   return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
               }else{
                   return ConsumeConcurrentlyStatus.RECONSUME_LATER;
               }
            }
        });
        CONSUMER.start();
    }

    @PreDestroy
    public void destory() {
        CONSUMER.shutdown();
    }
}
